package com.heima.article.stream;

import com.alibaba.fastjson.JSON;
import com.heima.model.article.mess.ArticleVisitStreamMess;
import com.heima.model.mess.UpdateArticleMess;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.ArrayUtils;
import org.apache.commons.lang3.StringUtils;
import org.apache.kafka.streams.KeyValue;
import org.apache.kafka.streams.StreamsBuilder;
import org.apache.kafka.streams.kstream.*;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.util.CollectionUtils;

import java.time.Duration;
import java.util.HashMap;
import java.util.Map;
import java.util.Set;

@Configuration
@Slf4j
public class KafkaStreamHelloListener {

    //1.设置窗口时间(多少秒进行合并发送)
    //2.分组
    //3 累加
    @Bean
    public KStream<String, String> kStream(StreamsBuilder streamsBuilder) {
        KStream<String, String> stream = streamsBuilder.stream("itcast-topic-in");
        stream.to("itcast-topic-out");

        stream.map((key, value) -> {
            //1 因为后面要根据articleId进行分组
            //先把articleId从JSON结构里面提取出来 放到key这个位置
            //为什么放在key  因为key这个位置目前这个阶段是没用的
            if (StringUtils.isEmpty(value)) {
                System.out.println("key" + key);
                System.out.println("value" + value);
                if (StringUtils.isEmpty(value)) {
                    log.warn("value为空");
                    return new KeyValue<>(key, value);
                }
                return new KeyValue<>(key, value);
            }
            UpdateArticleMess mess = JSON.parseObject(value, UpdateArticleMess.class);
            if (mess == null) {
                log.warn("value2为空");
                return new KeyValue<>(key, value);
            }
            Long articleId = mess.getArticleId();
            Integer add = mess.getAdd();
            UpdateArticleMess.UpdateArticleType type = mess.getType();

            return new KeyValue<>(articleId.toString(), type + ":" + add);
        }).groupBy((key, value) -> key)
                .windowedBy(TimeWindows.of(Duration.ofSeconds(5)))
                .aggregate(
                        //初始值 声明一个容器
                        new Initializer<String>() {
                            @Override
                            public String apply() {
                                return "COLLECTION:0,COMMENT:0,LIKES:0,VIEWS:0";
                            }
                        },
                        //累加
                        new Aggregator<String, String, String>() {
                            @Override
                            public String apply(String key, String value, String initData) {
                                //key 这个阶段 key 就是articleId
                                //value: type + add
                                //initData : COLLECTION:0,COMMENT:0,LIKES:0,VIEWS:0

                                //1 判断本次消息是什么类型
                                String[] valueArray = value.split(":");
                                //类型
                                String type = valueArray[0];
                                //数值
                                Integer add = Integer.valueOf(valueArray[1]);

                                //2 根据类型去更新对应的数值
                                //2.1 将字符串结构转换成Map结构 key为对应类型 value为类型房前的数值
                                Map<String, Integer> initMap = string2Map(initData);

                                //本次操作类型的
                                Integer totalAmount = initMap.get(type);
                                int  newAmount =  totalAmount + add;
                                //2.2 根据本次操作的type去map里面匹配数值 并且对数值进行+1 / -1 操作
                                initMap.put(type,newAmount);
                                //2.3 将map转回字符串结构
                                String result = map2String(initMap);
                                if (StringUtils.isEmpty(result)){
                                    log.warn("出现异常");
                                    return "COLLECTION:0,COMMENT:0,LIKES:0,VIEWS:0";
                                }
                                return result;
                            }
                        },
                        //合成器
                        Materialized.as("123")

                ).toStream()
                .map((key, value) -> {
                    return new KeyValue<>(key.key().toString(), string2JsonString(value, Long.parseLong(key.key())));
                })
                .to("itcast-topic-out");


        return stream;
    }

    /**
     * 字符串转实体类
     *
     * @param str       COLLECTION:0,COMMENT:0,LIKES:0,VIEWS:0
     * @param articleId 文章id
     * @return ArticleVisitStreamMess
     */
    private String string2JsonString(String str, Long articleId) {
        //1 收到字符串 COLLECTION:0,COMMENT:0,LIKES:0,VIEWS:0
        //2 解析字符串 -> 转成map结构
        Map<String, Integer> map = string2Map(str);

        ArticleVisitStreamMess mess = new ArticleVisitStreamMess();
        mess.setArticleId(articleId);
        //3 判断类型 根据不同的类型 累加到对应的实体类字段上
        Integer collection = map.get("COLLECTION");
        Integer comment = map.get("COMMENT");
        Integer likes = map.get("LIKES");
        Integer views = map.get("VIEWS");
        if (collection != null){
            mess.setCollect(collection);

        }
        if (comment != null){
            mess.setComment(comment);

        }
        if (likes != null){
            mess.setLike(map.get(likes));

        }
        if (views != null){
            mess.setView(map.get(views));

        }
        return JSON.toJSONString(mess);
    }


    /**
     * Map转字符串
     *
     * @param map map
     * @return COLLECTION:0,COMMENT:0,LIKES:0,VIEWS:0
     */
    private String map2String(Map<String, Integer> map) {
        if (CollectionUtils.isEmpty(map)){
            return null;
        }
        // 得到map里面所有的key
        Set<String> keySet = map.keySet();

        StringBuilder builder = new StringBuilder();
        //遍历key  获取他们对应的值
        for (String key : keySet) {
            builder.append(key).append(":").append(map.get(key));
            builder.append(",");
        }
        //StringBuild 将多个值组合起来
        //COLLECTION:0,COMMENT:0,LIKES:0,VIEWS:0
        String result = builder.toString();
        return result.substring(0, result.length() - 1);
    }

    /**
     * 字符串转Map结构
     *
     * @param str COLLECTION:0,COMMENT:0,LIKES:0,VIEWS:0
     * @return Map
     */
    private Map<String, Integer> string2Map(String str) {
        //1 结构COLLECTION:0,COMMENT:0,LIKES:0,VIEWS:0
        //2 判空
        if (StringUtils.isEmpty(str)){
            return new HashMap<>();
        }
        Map<String, Integer> map = new HashMap<>();
        //3先用 进行拆分 -> array
        String[] initDataArray = str.split(",");
        for (String data : initDataArray) {
            if (StringUtils.isEmpty(data)){
                continue;
            }
            //这个阶段得到的值是 COLLECTION:0
            String[] dataArray = data.split(":");
            if (ArrayUtils.isEmpty(dataArray)){
                continue;
            }
            // 获取对应的key和value
            String dataKey = dataArray[0];
            String dataValue = dataArray[1];

            map.put(dataKey, Integer.parseInt(dataValue));
        }

        return map;
    }


}
